package edu.wisc.cloudservice;

import java.util.HashMap;
import java.util.Map;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.google.protobuf.RpcCallback;
import com.google.protobuf.RpcController;

import edu.wisc.cloudservice.CloudManager.StartBrokerRequest;
import edu.wisc.cloudservice.CloudManager.StartBrokerResponse;
import edu.wisc.cloudservice.CloudManager.StartConsumerRequest;
import edu.wisc.cloudservice.CloudManager.StartConsumerResponse;
import edu.wisc.hevea.constants.ApplicationConstants;

import edu.wisc.hevea.launcher.BrokerLauncher;
import edu.wisc.hevea.launcher.ConsumerLauncher;

public class CloudServiceImpl implements CloudManager.CloudService.Interface {

	private static final Log logger = LogFactory.getLog(CloudServiceImpl.class);

	private Map<String,Integer> dcToLastStartedPort;
	private Map<String,Integer> dcToLastStartedConsumerPort;
	
	public CloudServiceImpl() {
		dcToLastStartedPort = new HashMap<String, Integer>();
		dcToLastStartedConsumerPort = new HashMap<String, Integer>();
		dcToLastStartedPort.put(ApplicationConstants.VIRGINIA_DC, 10005);
		dcToLastStartedPort.put(ApplicationConstants.IAD_DC, 20005);
		dcToLastStartedPort.put(ApplicationConstants.DUBLIN_DC, 30005);
		dcToLastStartedPort.put(ApplicationConstants.HONGKONG_DC, 40005);
		
		dcToLastStartedConsumerPort.put(ApplicationConstants.VIRGINIA_DC, 13005);
		dcToLastStartedConsumerPort.put(ApplicationConstants.IAD_DC, 23005);
		dcToLastStartedConsumerPort.put(ApplicationConstants.DUBLIN_DC, 33005);
		dcToLastStartedConsumerPort.put(ApplicationConstants.HONGKONG_DC, 43005);
		
	}

	@Override
	public void startBroker(RpcController controller,
			StartBrokerRequest request, RpcCallback<StartBrokerResponse> done) {

		String datacenter = request.getDataCenter();
		int numInstances = request.getNumInstances();
		
		int lastStartedPort = 10005;
		if(dcToLastStartedPort.containsKey(datacenter)) {
			lastStartedPort = dcToLastStartedPort.get(datacenter);
		} 
		
		for (int i = 0; i < numInstances; i++) {
			int portToStart = ++lastStartedPort; 
			logger.info(String
					.format("Starting %d th instance of broker in datacenter %s with port %d",
							i+1, datacenter, portToStart));
			
			BrokerLauncher bl = new BrokerLauncher("127.0.0.1", portToStart);
			new Thread(bl).start();
			dcToLastStartedPort.put(datacenter, portToStart);
		}
		
		// Fill code above
		StartBrokerResponse response = StartBrokerResponse.newBuilder()
				.setIsSuccess(true).build();
		done.run(response);

	}

	@Override
	public void startConsumer(RpcController controller,
			StartConsumerRequest request,
			RpcCallback<StartConsumerResponse> done) {
		String datacenter = request.getDataCenter();
		int numInstances = request.getNumInstances();
		
		int lastStartedPort = 13005;
		if(dcToLastStartedConsumerPort.containsKey(datacenter)) {
			lastStartedPort = dcToLastStartedConsumerPort.get(datacenter);
		} 
		
		for (int i = 0; i < numInstances; i++) {
			int portToStart = ++lastStartedPort; 
			logger.info(String
					.format("Starting %d th instance of consumer in datacenter %s with port %d",
							i+1, datacenter, portToStart));
			ConsumerLauncher cl = new ConsumerLauncher("127.0.0.1", portToStart);
			new Thread(cl).start();			
			dcToLastStartedConsumerPort.put(datacenter, portToStart);
		}
		
		// Fill code above
		StartConsumerResponse response = StartConsumerResponse.newBuilder()
				.setIsSuccess(true).build();
		
		done.run(response);
		
	}

}
